<?php

namespace app\common\service\SystemRabbitMqService;

use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;
use app\BaseService;
use app\Normal;

abstract class AmqpMessageManager extends BaseService
{
    /**
     * @var object
     */
    protected $connection;
    
    /**
     * 队列
     * @var string
     */
    protected $queueName;
    
    /**
     * 路由
     * @var string
     */
    protected $routingKey;
    
    /**
     * 交换机
     * @var string
     */
    protected $exchangeName;
    
    /**
     * 预消费数据量
     * @var int
     */
    protected $prefetch_count = 10;
    
    /**
     * 服务端接收生成消息应答超时时间
     * @var int
     */
    protected $timeout = 3;
    
    /**
     * 通道
     * @var object
     */
    protected $chanel;
    
    /**
     * 是否需要消息应答确认
     * @var bool
     */
    protected $isConfirm = true;
    
    /**
     * 是否需要重回队列
     * @var bool
     */
    protected $isReject = true;
    
    /**
     * 消息者标识名称
     * @var bool
     */
    protected $consumer_title = null;
    public function __construct( $mq_name = null )
    {
        $config = config('amqp');
        if( empty($config) ) {
            throw new \Exception("Amqp config Is Not Readying,Please check it!");
        }
        
        $this->setQueueName($mq_name);
        $this->setRoutingKey($mq_name);
        $this->setExchangeName($mq_name);
                
        //初始化连接参数
        $this->connection = new AMQPStreamConnection(
            $config['host'],
            $config['port'],
            $config['username'],
            $config['password'],
            '/',
            false,
            'AMQPLAIN',
            null,
            'en_US',
            120.0,
            120.0,
            null,
            true,
            30
            );
        /**
         * 2,获取信道
         * $channel_id ==> 信道id，不传则获取$channel[""]信道，每个信道相互独立
         * 再无 => 则循环$this->channle数组，下标从1到最大信道数找第一个不是AMQPChannel对象的下标，实例化并返回AMQPChannel对象，
         * 无则抛出异常No free channel ids
         */
        $this->chanel = $this->connection->channel();
        
        // 声明交换机
        // 在信道里创建交换机，未显式声明交换机都是使用匿名交换机
        $this->chanel->exchange_declare(
            $this->exchangeName, // 交换机名字
            'direct',            // 交换机类型，主题交换机，集合对应关系：N：1
            false,
            true, // 交换机将在服务器重启后继续存在
            false // 通道关闭的时候，交换机不会被删除
            );
        
        // 4，声明消费者队列
        $this->chanel->queue_declare(
            $this->queueName, // 队列名称
            false,
            true, // 队列是否持久化， 持久化队列为true
            false, // 当前连接不在时，队列是否自动删除
            false // 没有consumer时，队列是否自动删除
            );
        
        /**
         * 绑定队列到交换机，当未显示绑定交换机时，默认是绑定匿名交换机
         * 绑定：交换机与队列的关系，如下面这句代码意思是，$queue_name队列对$exchange_name交换机数据感兴趣，
         * 该队列就消费该交换机传过来的数据：这个队列（queue）对这个交换机（exchange）的消息感兴趣. $routing_key默认为空，表示对该交换机所有消息感兴趣，
         * 如果值不为空，则该队列只对该类型的消息感兴趣（除了fanout交换机以外）
         */
        $this->chanel->queue_bind($this->queueName,$this->exchangeName,$this->routingKey);
        
        //加上这个就好了 这个10 就是Unacked 里面的值，表示预先取出多少值来消费  prefetch_count
        $this->chanel->basic_qos(null, $this->prefetch_count, null);
        
    }
    
    /**
     * 队列
     * @var string
     */
    public function setQueueName( string $QueueName = null )
    {
        if( $QueueName ) $this->queueName = $QueueName;
    }
    
    /**
     * 路由
     * @var string
     */
    public function setRoutingKey( string $RoutingKey = null )
    {
        if( $RoutingKey ) $this->routingKey = $RoutingKey;
    }
    
    /**
     * 交换机
     * @var string
     */
    public function setExchangeName( string $ExchangeName = null )
    {
        if( $ExchangeName ) $this->exchangeName = $ExchangeName;
    }
    
    /**
     * 获取队列
     * @var string
     */
    public function getQueueName()
    {
        return $this->queueName;
    }
    
    /**
     * 获取路由
     * @var string
     */
    public function getRoutingKey()
    {
        return $this->routingKey;
    }
    
    /**
     * 获取交换机
     * @var string
     */
    public function getExchangeName()
    {
        return $this->exchangeName;
    }
    
    /**
     * 发布消息(支持单个发送，批量发送【批量发送$isMultiMessage为true】)
     * @param array $message
     * @param boolean $isMultiMessage
     * @return mixed
     */
    public function produce( array $message ,bool $isMultiMessage = false )
    {
        if( $this->isConfirm ) {
            $result = false;
            $this->chanel->confirm_select();
            $this->chanel->set_ack_handler(function () use (&$result) {
                $result = true;
            });
        } else {
            $result = true;
        }
        
        try {
            
            if( $isMultiMessage ) 
            {
                foreach ($message as $item ) 
                {
                    $contents = json_encode($item,JSON_UNESCAPED_UNICODE);
                    $msg = new AMQPMessage($contents);
                    $this->chanel->basic_publish($msg,false,$this->queueName);
                }
            } else {
                $contents = json_encode($message,JSON_UNESCAPED_UNICODE);
                $msg = new AMQPMessage($contents);
                $this->chanel->basic_publish($msg,false,$this->queueName);
            }
        
            //如果要确认的话，需要进行等待应答消息接收结果
            if( $this->isConfirm ) {
                $this->chanel->wait_for_pending_acks_returns($this->timeout);
            }
            
            //使用完成之后关闭连接
            $this->chanel->close();
            $this->connection->close();
            if( $result ) {
                return Normal::returnCode(Normal::SUC,"发布成功");
            }
            return Normal::returnCode(Normal::FAIL,"发布失败");
        } catch ( \Exception $e ) {
            $this->chanel->close();
            $this->connection->close();
            return Normal::returnCode(Normal::FAIL,"发布消息失败:【{$e->getMessage()}】");
        }
    }
    
    /**
     * 消费者
     */
    public function consumer()
    {
        $callback = function($msg)
        {
            $message = json_decode($msg->body,true);
            $result = $this->messageConsumer($message);
            //处理成功则应答OK
            if( $result['code'] == Normal::SUC ) 
            {
                $msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);
            } 
            else 
            {
                //需要等待消息，并且重回消息队列数据
                if( $this->isReject )
                {
                    $msg->delivery_info['channel']->basic_reject($msg->delivery_info['delivery_tag'],true);
                    sleep(10);
                } 
                else
                {
                    $msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);
                }
            }
        };
        
        //监听消息接收
        $this->chanel->basic_consume($this->queueName, '', false, false, false, false, $callback);
        while(count($this->chanel->callbacks)) {
            $this->chanel->wait();
        }
    }
}